#include "StdAfx.h"
#include "tcp_syn_packet_process.h"

tcp_syn_packet_sender::tcp_syn_packet_sender(void)
:notifier_(0,0,0)
{
	queue_.high_water_mark(1024*1024/*(aot::inet::PKT_MAX_LEN+aot::inet::PKT_HEADER_LEN)*10*/);
}

tcp_syn_packet_sender::~tcp_syn_packet_sender(void)
{
}

int 
tcp_syn_packet_sender::open(ACE_Event_Handler* eh)
{
	this->eh_ = eh;

	this->notifier_.event_handler(eh);
	this->notifier_.reactor(eh->reactor());
	this->notifier_.mask(ACE_Event_Handler::WRITE_MASK);

	this->queue_.notification_strategy (&this->notifier_);
	return 0;
}

int 
tcp_syn_packet_sender::putq(aot::inet::ipacket* pkt, const aot::inet::inet_time_t* tm_out/* = NULL*/)
{
	int r;
	ACE_Message_Block* mb = (ACE_Message_Block*)pkt->header_impl();
	bool has_header = true;

	if( mb )
	{
		mb->cont((ACE_Message_Block*)pkt->body_impl());
	}
	else
	{
		has_header = false;
		mb = (ACE_Message_Block*)pkt->body_impl();
	}

#define __REUSE_PKT__()			\
	if( has_header ){				\
	pkt->body_impl(mb->cont());	\
	mb->cont(NULL);				\
	pkt->header_impl(mb);		\
	}								\
	else{							\
	pkt->body_impl(mb);			\
	}								\

#define __UNUSE_PKT__()			\
	pkt->header_impl(NULL);		\
	pkt->body_impl(NULL);		\
	pkt->destroy();				\

	if( tm_out )
	{
		ACE_Time_Value t = ACE_OS::gettimeofday();
		t += ACE_Time_Value(tm_out->sec, tm_out->usec);

		r = this->queue_.enqueue_tail(mb, &t);

		if( r == -1 )
		{
			__REUSE_PKT__();

			if( ACE_OS::last_error() == EWOULDBLOCK )
				return aot::inet::e_ret_again;
			return aot::inet::e_ret_failed;
		}
	}
	else
	{
		r = this->queue_.enqueue(mb);

		if( r == -1 )
		{
			__REUSE_PKT__();
			return aot::inet::e_ret_failed;
		}
	}

	__UNUSE_PKT__();
	return aot::inet::e_ret_ok;
}

ACE_Message_Queue<ACE_MT_SYNCH>* 
tcp_syn_packet_sender::get_msg_queue()
{
	return &this->queue_;
}

int 
tcp_syn_packet_sender::mb_to_iovec(iovec v[], int num, ACE_Message_Block* mb)
{
	int i;
	int n = num;
	ACE_Message_Block* p = mb;
	for( i = 0; NULL != p && i < n; p = p->cont())
	{
		if( p->length() == 0 )
			continue;
		v[i].iov_base = p->rd_ptr();
		v[i].iov_len = p->length();
		++i;
	}
	return i;
}

size_t 
tcp_syn_packet_sender::update_mb(size_t send_len, ACE_Message_Block* mb)
{
	ACE_Message_Block* p = mb;
	for( ; NULL != p && send_len > 0; p = p->cont() )
	{
		if( p->length() == 0 )
			continue;
		if( send_len > p->length() )
		{
			send_len -= p->length();
			p->rd_ptr(p->length());
			continue;
		}
		else
		{
			p->rd_ptr(send_len);
			return 0;
		}
	}
	return send_len;
}

int tcp_syn_packet_sender::send(ACE_HANDLE h)
{
	ACE_SOCK_Stream peer;
	if( h == ACE_INVALID_HANDLE )
	{
		return -1;
	}
	
	peer.set_handle(h);

	ACE_Message_Block *mb;
	ACE_Time_Value nowait (ACE_OS::gettimeofday ());

	int iovec_num = 0;
	const int MAX_IOVEC_NUM = 2;

	while ( -1 != this->queue_.dequeue_head (mb, &nowait) )
	{
		iovec v[MAX_IOVEC_NUM];
		iovec_num = mb_to_iovec(v, MAX_IOVEC_NUM, mb);

		if( iovec_num == 0 )
		{
			mb->release();
			continue;
		}

		ssize_t send_cnt = peer.sendv(v, iovec_num);
		if ( send_cnt == -1 )
		{
			if( ACE_OS::last_error() == EWOULDBLOCK )
			{
				this->queue_.enqueue_head (mb);
				break;
			}
			else
			{
				mb->release();
				aot_log_error(AOT_LM_ERROR,  "send failed");
				return -1;
			}
		}
		else
		{
			update_mb(send_cnt, mb);
		}

		if ( mb->total_length() > 0 )
		{
			this->queue_.enqueue_head (mb);
			break;
		}
		else
		{
			mb->release();
		}
	}

	{
		ACE_Guard<ACE_MT_SYNCH::MUTEX> __mon(this->queue_.lock());

		if (this->queue_.is_empty ())
			this->eh_->reactor ()->cancel_wakeup(this->eh_, ACE_Event_Handler::WRITE_MASK);
		else
			this->eh_->reactor ()->schedule_wakeup(this->eh_, ACE_Event_Handler::WRITE_MASK);
	}

	return 0;
}

/* - * - */

tcp_syn_packet_recver::tcp_syn_packet_recver()
{
	this->pkt_ = NULL;
	this->queue_.high_water_mark(1024*1024/*(aot::inet::PKT_MAX_LEN+aot::inet::PKT_HEADER_LEN)*10*/);
}

tcp_syn_packet_recver::~tcp_syn_packet_recver(void)
{
	if(this->pkt_)
	{
		delete this->pkt_;
		this->pkt_ = NULL;
	}
}

int 
tcp_syn_packet_recver::open()
{
	return 0;
}

int 
tcp_syn_packet_recver::getq(aot::inet::ipacket** pkt, const aot::inet::inet_time_t* tm_out)
{
	ACE_Message_Block* mb = NULL;
	int r;
	if( tm_out )
	{
		ACE_Time_Value t = ACE_OS::gettimeofday();
		t += ACE_Time_Value(tm_out->sec, tm_out->usec);

		r = this->queue_.dequeue(mb, &t);

		if( r == -1 )
		{
			if( ACE_OS::last_error() == EWOULDBLOCK )
				return aot::inet::e_ret_again;
			return aot::inet::e_ret_failed;
		}
	}
	else
	{
		r = this->queue_.dequeue(mb);

		if( r == -1 )
		{
			return aot::inet::e_ret_failed;
		}
	}

	if( NULL == mb )
		return aot::inet::e_ret_failed;

	packet_impl* p = packet_impl::create_packet();
	if( NULL == p )
	{
		mb->release();
		return aot::inet::e_ret_failed;
	}

	*pkt = (aot::inet::ipacket*)p;
	p->header_ = mb;
	p->body_ = mb->cont();
	mb->cont(NULL);
	return aot::inet::e_ret_ok;
}

bool 
tcp_syn_packet_recver::process_packet(tcp_syn_packet_sender* pkt_sender)
{
	aot::inet::packet_header h(this->pkt_->header_base());

	switch( h.type() )
	{
	case aot::inet::e_pkt_type_echo:
		{
			h.type(aot::inet::e_pkt_type_echo_reply);
			aot::inet::inet_time_t tm;
			tm.sec = 0; tm.usec = 0;
			if( aot::inet::e_ret_ok != pkt_sender->putq(this->pkt_, &tm) )
			{
				this->pkt_->destroy();
				this->pkt_ = NULL;
			}
			else
			{
				this->pkt_ = NULL;
			}
			return true;
		}
		break;
	case aot::inet::e_pkt_type_echo_reply:
		{
			this->pkt_->destroy();
			this->pkt_ = NULL;
			return true;
		}
		break;
	case aot::inet::e_pkt_type_heartbeat:
		{
			this->pkt_->destroy();
			this->pkt_ = NULL;
			return true;
		}
		break;
	case aot::inet::e_pkt_type_data:
	default:
		{
			ACE_Message_Block* mb = this->pkt_->header_;
			mb->cont(this->pkt_->body_);
			if( -1 == this->queue_.enqueue(mb) )
			{
				this->pkt_->destroy();
				this->pkt_ = NULL;
				return false;
			}
			this->pkt_->header_ = NULL;
			this->pkt_->body_ = NULL;
			this->pkt_->destroy();
			this->pkt_ = NULL;

			return true;
		}
		break;
	}

	return true;
}

int 
tcp_syn_packet_recver::recv(ACE_HANDLE h, tcp_syn_packet_sender* pkt_sender)
{
	int r;
	while(aot::inet::e_ret_ok == (r = recv_i(h)) )
	{
		if( !process_packet(pkt_sender) )
			return -1;
	}

	return r == aot::inet::e_ret_failed ? -1 : 0;
}

int 
tcp_syn_packet_recver::recv_i(ACE_HANDLE h)
{
	ACE_SOCK_Stream peer(h);
	int r = 0;
	bool is_get_header = false;
	aot_uint32_t exh_len = 0;
	aot_uint32_t body_len = 0;

	if( NULL == this->pkt_ )
	{
		this->pkt_ = packet_impl::create_packet();
		if( NULL == this->pkt_ )
			goto do_failed;
		if( !this->pkt_->create_header() )
			goto do_failed;
	}

	is_get_header = (this->pkt_->header_->space() == 0);

	for(;;)
	{
		if( !is_get_header )
		{
			r = peer.recv(this->pkt_->header_->wr_ptr(), this->pkt_->header_->space());
		}
		else 
		{
			if( NULL == this->pkt_->body_ )
			{
				aot::inet::packet_header ph( this->pkt_->header_->base() );

				if( !ph.is_valid() )
					goto do_failed;

				body_len = ph.data_len();
				exh_len = ph.extern_header_len();
				if( body_len + exh_len == 0)
					goto do_ok;

				if( !this->pkt_->create_body(body_len + exh_len) )
					goto do_failed;
			}

			r = peer.recv(this->pkt_->body_->wr_ptr(), this->pkt_->body_->space());
		}

		if( 0 == r)
		{
			aot_log_error(AOT_LM_ERROR, "recv EOF, peer has closed");
			goto do_failed;
		}

		if( -1 == r )
		{
			if(ACE_OS::last_error() == EWOULDBLOCK)
			{
				goto do_again;
			}
			else
			{
				aot_log_error(AOT_LM_ERROR, "recv -1, peer has disconnect");
				goto do_failed;
			}
		}

		if( !is_get_header )
		{
			this->pkt_->header_->wr_ptr(r);
			is_get_header = (this->pkt_->header_->space() == 0);
		}
		else
		{
			this->pkt_->body_->wr_ptr(r);
			if( this->pkt_->body_->space() == 0 )
				goto do_ok;
		}

	}

do_ok:
	return aot::inet::e_ret_ok;

do_failed:
	if( this->pkt_ )
	{
		this->pkt_->destroy();
		this->pkt_ = NULL;
	}
	return aot::inet::e_ret_failed;

do_again:
	return aot::inet::e_ret_again;
}
